package com.matrix.draft.sockettest.day11;

import io.netty.buffer.ByteBufAllocator;
import io.netty.channel.*;
import io.netty.channel.embedded.EmbeddedChannel;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.handler.logging.LoggingHandler;
import io.netty.util.concurrent.DefaultPromise;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;
import lombok.extern.slf4j.Slf4j;

import java.nio.charset.StandardCharsets;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

@Slf4j
public class TestFuture {
    public static void main(String[] args) throws Exception {
        test4();
    }

    // 测试Jdk Future
    public static void test1() throws Exception {
        ExecutorService executorService = Executors.newFixedThreadPool(2);
        Future<Integer> future = (Future<Integer>) executorService.submit(() -> {
            Thread.sleep(1000);
            System.out.println("计算完成");
            return 1;
        });
        Thread.sleep(3000);
        System.out.println(future.get());
        executorService.shutdown();
    }

    // 测试Netty Future
    public static void test2() throws Exception {
        NioEventLoopGroup eventLoopGroup = new NioEventLoopGroup(1);
        EventLoop eventLoop = eventLoopGroup.next();
        io.netty.util.concurrent.Future<Integer> nettyFuture = eventLoop.submit(() -> {
            Thread.sleep(1000);
            log.debug("计算完成");
            return 1;
        });
        nettyFuture.addListener(new GenericFutureListener<Future<? super Integer>>() {
            @Override
            public void operationComplete(Future<? super Integer> future) throws Exception {
                if (future.isSuccess()) {
                    log.debug("最终结果: {}", future.getNow());
                }
            }
        });
    }

    // 测试Promise
    public static void test3() throws Exception {
        // 准备Promise对象(存放结果的容器)
        EventLoop eventLoop = new NioEventLoopGroup(1).next();
        DefaultPromise<Integer> promise = new DefaultPromise<>(eventLoop);
        // 交由其他线程来处理
        new Thread(() -> {
            try {
                log.debug("开始计算...");
                Thread.sleep(1000);
                promise.setSuccess(888); // 表示任务完成,sync()会取消阻塞
            } catch (InterruptedException e) {
                e.printStackTrace();
                promise.setFailure(e); // 表示任务失败, 抛出异常.
            }
        }).start();
        promise.sync();
        log.debug("最总结果: {}", promise.getNow());
        eventLoop.shutdownGracefully();
    }

    // 测试EmbeddedChannel
    public static void test4() throws Exception {
        EmbeddedChannel embeddedChannel = new EmbeddedChannel(
                // LoggingHandler
                new LoggingHandler(),
                // in1
                new ChannelInboundHandlerAdapter() {
                    @Override
                    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                        log.debug("in1");
                        ctx.fireChannelRead(msg);
                    }
                },
                // in2
                new ChannelInboundHandlerAdapter() {
                    @Override
                    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                        log.debug("in2");
                        ctx.fireChannelRead(msg);
                    }
                },
                // out3
                new ChannelOutboundHandlerAdapter() {
                    @Override
                    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
                        log.debug("out3");
                        ctx.write(msg);
                    }
                },
                // out4
                new ChannelOutboundHandlerAdapter() {
                    @Override
                    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
                        log.debug("out4");
                        ctx.write(msg);
                    }
                }
        );
        // 模拟入站 (假设当前是客户端,而服务端向我们这里写入)
        embeddedChannel
                .writeInbound(ByteBufAllocator.DEFAULT.buffer().writeBytes("Hello EmbeddedChannel".getBytes(StandardCharsets.UTF_8)));
        // 模拟出站 (假设当前是客户端,而我们向服务端写入)
        embeddedChannel
                .writeOutbound(ByteBufAllocator.DEFAULT.buffer().writeBytes("Hello EmbeddedChannel".getBytes(StandardCharsets.UTF_8)));
        embeddedChannel.closeFuture().addListener(future -> {
            if(future.isSuccess()){
                log.debug("通道关闭成功");
            }
        });
    }

}
